package com.tl.job014.utils.kafka;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

/**
 * kafka 消费者工具类
 * 
 * @author yingfing
 * @date 2021年1月20日 @time 下午10:50:22
 */
public class KafkaConsumerUtil4Java {

	public KafkaConsumer<String, String> consumer = null;

	public KafkaConsumerUtil4Java(String bootStrapServer, String topic) {
		// 通过Properties工具类封装kafka自提供KafkaProducer的参数
		Properties prop = new Properties();

		prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServer);
		prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
				StringDeserializer.class.getName());
		prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
				StringDeserializer.class.getName());

		// 设置 group id
		prop.put(ConsumerConfig.GROUP_ID_CONFIG, "Group4Job014Byzyf");

		// 设置是否自动提交
		prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);

		// 设置自动提交的时间间隔
		prop.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);

		// 初始化 kafka 自提供的工具类，并做订阅操作
		consumer = new KafkaConsumer<String, String>(prop);
		consumer.subscribe(Arrays.asList(topic));
	}

	public static void main(String[] args) {

		// 1、传入bootstrap-server参数
		String bootStrapServer = "cluster1.hadoop:6667,clusterkk::6667";
		String topic = "zyfKafkaTest";

		// 2、构建KafkaProducerUtil4Java对象
		KafkaConsumerUtil4Java consumerUtil = new KafkaConsumerUtil4Java(
				bootStrapServer, topic);

		// 3、消费生产消息发送到指定的topic
		boolean isRunningFlag = true;
		while (isRunningFlag) {
			ConsumerRecords<String, String> records = consumerUtil.consumer
					.poll(1000);

			for (ConsumerRecord<String, String> record : records) {
				System.out.printf("key=%s,offset=%d,value=%s", record.key(),
						record.offset(), record.value());
				System.out.println();
			}
		}

		// 4、关闭链接
		consumerUtil.consumer.close();
	}
}
